package top.wintp.offlinedataanalysis.util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Map;

import javax.jws.Oneway;


/**
 * description:
 * <p>
 * author:  upuptop
 * <p>
 * qq: 337081267
 * <p>
 * CSDN:   http://blog.csdn.net/pyfysf
 * <p>
 * cnblogs:   http://www.cnblogs.com/upuptop
 * <p>
 * blog:   http://wintp.top
 * <p>
 * email:  pyfysf@163.com
 * <p>
 * time: 2019/08/2019/8/16
 * <p>
 */
public class HBaseUtils {

    /**
     * 获取连接
     *
     * @param config
     * @return
     * @throws IOException
     */
    public static Connection getConn(Configuration config) throws IOException {
        return ConnectionFactory.createConnection(config);
    }

    /**
     * 插入数据
     *
     * @param config
     * @param tableName
     * @param familyName
     * @param dataMap
     * @throws IOException
     */
    public static void insert(Configuration config, String tableName, String familyName, Map<String, String> dataMap) throws IOException {
        Connection conn = getConn(config);
        Table table = conn.getTable(TableName.valueOf(tableName));
        Put put = new Put(Bytes.toBytes(familyName));

        //添加数据
        for (Map.Entry<String, String> data : dataMap.entrySet()) {
            put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(data.getKey()), Bytes.toBytes(data.getValue()));
        }

        table.put(put);


        table.close();
        conn.close();
    }

    /**
     * 创建命名空间
     *
     * @param config
     * @param nameSpace
     * @throws IOException
     */
    public static void createNameSpace(Configuration config, String nameSpace, Map<String, String> nameSpaceConf) throws IOException {
        Connection conn = getConn(config);
        Admin admin = conn.getAdmin();

        //创建命名空间
        NamespaceDescriptor nameSpaceDesc = NamespaceDescriptor.create(nameSpace).addConfiguration(nameSpaceConf).build();
        admin.createNamespace(nameSpaceDesc);

        admin.close();
        conn.close();
    }

    /**
     * 创建表
     *
     * @param config
     * @param tableName
     * @param familyName
     * @throws IOException
     */
    public static void createTable(Configuration config, String tableName, String familyName) throws IOException {
        Connection conn = getConn(config);
        Admin admin = conn.getAdmin();

        //创建表
        HTableDescriptor hTableDesc = new HTableDescriptor(TableName.valueOf(tableName));
        HColumnDescriptor hColumnDesc = new HColumnDescriptor(Bytes.toBytes(familyName));
        hTableDesc.addFamily(hColumnDesc);
        admin.createTable(hTableDesc);


        admin.close();
        conn.close();
    }


}
